package cn.lyjuan.first.hadoop.util;

import cn.lyjuan.base.util.RandomUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.*;
import java.net.URI;
import java.net.URISyntaxException;

/**
 * HDFS工具类
 */
public class HDFSUtil {
    public static final String HADOOP_URI = "hdfs://192.168.1.200:9000";
    public static final String USER = "root";
    public static Configuration CONF = new Configuration();
    public static FileSystem FS;

    static {
        try {
            loadFS();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private static void loadFS() throws URISyntaxException, IOException, InterruptedException {

        if (null != FS) return;
        synchronized (HDFSUtil.class) {
            if (null != FS) return;
            URI uri = new URI(HADOOP_URI);
            CONF = new Configuration();
            // 单机hadoop情况下，namenode返回datanode的hostname和私有IP地址，只能使用hostname连接，并且需要在本机的/etc/hosts上配置正确的hostname解析
            CONF.set("dfs.client.use.datanode.hostname", "true");

            FS = FileSystem.get(uri, CONF, USER);
            System.out.println("home ==> " + FS.getHomeDirectory());
        }
    }

    /**
     * 删除
     *
     * @param path
     * @return
     */
    public static boolean del(Path path) {
        try {
            return FS.delete(path, true);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 删除，关联{@link HDFSUtil#del(Path)}
     *
     * @param path
     * @return
     */
    public static boolean del(String path) {
        return del(new Path(path));
    }

    public static FileStatus[] ls(String path) {
        return ls(new Path(path));
    }

    /**
     * 列出目录下所有的文件
     *
     * @param path
     * @return
     */
    public static FileStatus[] ls(Path path) {
        try {
            FileStatus[] files = FS.listStatus(path);
            return files;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 写入本地文件
     *
     * @param localFlie
     * @param context
     */
    public static void writeLocal(String localFlie, String context) {
        File file = new File(localFlie);
        if (!file.getParentFile().isDirectory())
            file.getParentFile().mkdirs();

        BufferedOutputStream out = null;
        try {
            out = new BufferedOutputStream(new FileOutputStream(file));
            out.write(context.getBytes());
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            if (null != out) {
                try {
                    out.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 写入内容到HDFS远程文件
     *  @param remotePath
     * @param context
     */
    public static void writeRemote(Path remotePath, String context) {
        String file = FileUtil.class.getResource("./").getFile();
        file += "/" + RandomUtils.uuid();
        writeLocal(file, context);

        try {
            FS.copyFromLocalFile(new Path(file), remotePath);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public static byte[] readRemote(String remotePath) {
        return readRemote(new Path(remotePath));
    }

    /**
     * 读取远程文件内容
     *
     * @param remotePath
     * @return
     */
    public static byte[] readRemote(Path remotePath) {
        FSDataInputStream in = null;
        try {
            in = FS.open(remotePath);
            ByteArrayOutputStream out = new ByteArrayOutputStream();
            int len = -1;
            byte[] b = new byte[10 * 1024];
            while (-1 != (len = in.read(b))) {
                out.write(b, 0, len);
            }

            return out.toByteArray();
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            if (null != in) {
                try {
                    in.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 获取HDFS文件系统工具类
     *
     * @return
     */
    public static FileSystem fs() {
        return FS;
    }

    public static Configuration conf() {
        return CONF;
    }
}
